package server;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.nio.ByteBuffer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

public class ServerRecvThread extends Thread {

	private DatagramSocket sock;
	
	/* PubIds -> last sequence number we've seen */
	private ConcurrentHashMap<String, Long> lastSeq;
	/* User queue */
	private BlockingQueue<Message> uOutgoing;
	/* Server queue */
	private BlockingQueue<Message> sOutgoing;
	
	public ServerRecvThread(BlockingQueue<Message> uOutgoing, BlockingQueue<Message> sOutgoing, DatagramSocket sSock, ConcurrentHashMap<String, Long> lastSeq) {
		super("ServRecv-" + sSock.getLocalPort());
		this.sock = sSock;
		this.uOutgoing = uOutgoing;
		this.sOutgoing = sOutgoing;
		
		this.lastSeq = lastSeq;
	}
	
	public void log(String msg) {
		Server.log(this.getName(), msg);
	}

	@Override
	public void run() {
		while (true) {
			byte[] buf = new byte[256];
			DatagramPacket pac = new DatagramPacket(buf, buf.length);
			try {
				sock.receive(pac);
				Message m = Message.deserialize(ByteBuffer.wrap(pac.getData()));
				
				/* Check if we've done this already */
				Long oldSeq = lastSeq.get(m.pubId);
				if (oldSeq == null || m.seq > oldSeq) {
					lastSeq.put(m.pubId, m.seq);
					uOutgoing.put(m); // clients
					sOutgoing.put(m); // servers
				}
			} catch (IOException e) {
				e.printStackTrace();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

}
